package com.seedog.mq.example;

import com.seedog.mq.Consumer;
import com.seedog.mq.MessageQueueFactory;
import com.seedog.mq.Producer;
import com.seedog.mq.exception.TopicNotFoundException;
import com.seedog.mq.model.Message;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 消息队列使用示例类，演示如何使用SeeDog消息队列中间件
 */
public class MessageQueueExample {
    
    public static void main(String[] args) throws Exception {
        // 示例主题名称
        String topicName = "test-topic";
        
        // 创建主题（实际使用中，主题通常由管理员提前创建）
        // 这里我们通过MessageQueueFactory获取消息队列实例并创建主题
        MessageQueueFactory.getInstance().getMessageQueue().createTopic(topicName);
        System.out.println("主题 " + topicName + " 创建成功");
        
        // 创建生产者
        Producer producer = MessageQueueFactory.getInstance().createProducer();
        System.out.println("生产者创建成功，ID: " + producer.getProducerId());
        
        // 创建消费者
        Consumer consumer = MessageQueueFactory.getInstance().createConsumer();
        System.out.println("消费者创建成功，ID: " + consumer.getConsumerId());
        
        // 消费者订阅主题
        consumer.subscribe(topicName);
        System.out.println("消费者已订阅主题: " + topicName);
        
        // 生产者发送消息
        sendMessagesExample(producer, topicName);
        
        // 消费者接收消息
        receiveMessagesExample(consumer, topicName);
        
        // 发送带过期时间的消息
        sendExpiringMessagesExample(producer, topicName);
        
        // 批量发送消息
        sendBatchMessagesExample(producer, topicName);
        
        // 再次接收消息
        receiveMessagesExample(consumer, topicName);
        
        // 清理资源
        cleanupResources(producer, consumer, topicName);
    }
    
    /**
     * 演示发送消息
     */
    private static void sendMessagesExample(Producer producer, String topicName) throws TopicNotFoundException {
        // 发送文本消息
        String messageId = producer.send(topicName, "Hello, SeeDog Message Queue!");
        System.out.println("已发送文本消息到主题: " + topicName + ", 消息ID: " + messageId);
        
        // 创建并发送自定义消息
        Message customMessage = new Message(topicName, "This is a custom message with properties".getBytes());
        customMessage.getProperties().setHeader("priority", "high");
        customMessage.getProperties().setHeader("sender", producer.getProducerId());
        
        String customMessageId = producer.send(topicName, customMessage);
        System.out.println("已发送自定义消息到主题: " + topicName + ", 消息ID: " + customMessageId);
    }
    
    /**
     * 演示发送带过期时间的消息
     */
    private static void sendExpiringMessagesExample(Producer producer, String topicName) throws TopicNotFoundException {
        // 直接使用新添加的方法发送带过期时间的文本消息（5秒后过期）
        String messageId = producer.send(topicName, "This message will expire after 5 seconds", 5000);
        System.out.println("已发送带过期时间的消息到主题: " + topicName + ", 消息ID: " + messageId);
        
        // 演示发送带过期时间的另一种方式
        Message expiringMessage = new Message(topicName, "Another expiring message".getBytes());
        expiringMessage.setExpireTime(5000);
        expiringMessage.getProperties().setHeader("type", "expiring");
        
        String expiringMessageId = producer.send(topicName, expiringMessage);
        System.out.println("已发送第二条带过期时间的消息到主题: " + topicName + ", 消息ID: " + expiringMessageId);
    }
    
    /**
     * 演示批量发送消息
     */
    private static void sendBatchMessagesExample(Producer producer, String topicName) throws TopicNotFoundException {
        System.out.println("开始批量发送消息...");
        
        List<Message> messages = new ArrayList<>();
        for (int i = 1; i <= 5; i++) {
            Message batchMessage = new Message(topicName, ("Batch message #" + i).getBytes());
            batchMessage.getProperties().setHeader("batch_id", "batch-001");
            batchMessage.getProperties().setHeader("sequence", String.valueOf(i));
            messages.add(batchMessage);
        }
        
        List<String> messageIds = producer.sendBatch(topicName, messages);
        System.out.println("批量发送消息完成，共发送 " + messageIds.size() + " 条消息");
        System.out.println("消息ID列表: " + messageIds);
    }
    
    /**
     * 演示接收消息
     */
    private static void receiveMessagesExample(Consumer consumer, String topicName) {
        System.out.println("开始接收消息...");
        
        int receivedCount = 0;
        int maxMessages = 10; // 最多接收10条消息
        
        while (receivedCount < maxMessages) {
            try {
                // 尝试从队列中获取消息，最多等待2秒
                Message message = consumer.poll(topicName, 2000);
                
                if (message != null) {
                    receivedCount++;
                    System.out.println("收到消息 #" + receivedCount + ": " + new String(message.getBody()));
                    System.out.println("消息ID: " + message.getMessageId());
                    System.out.println("消息主题: " + message.getTopic());
                    
                    // 检查消息是否已过期
                    if (message.isExpired()) {
                        System.out.println("注意：此消息已过期");
                    }
                    
                    // 打印消息属性
                    if (message.getProperties() != null && !message.getProperties().getHeaders().isEmpty()) {
                        System.out.println("消息属性: " + message.getProperties().getHeaders());
                    }
                    
                    // 确认消息已消费
                    boolean acknowledged = consumer.acknowledge(topicName, message.getMessageId());
                    if (acknowledged) {
                        System.out.println("消息已确认消费: " + message.getMessageId());
                    } else {
                        System.out.println("消息确认失败: " + message.getMessageId());
                    }
                    System.out.println("----------------------------------------");
                } else {
                    // 没有收到消息，退出循环
                    break;
                }
            } catch (Exception e) {
                System.err.println("接收消息时发生错误: " + e.getMessage());
                e.printStackTrace();
                break;
            }
        }
        
        System.out.println("接收消息完成，共收到 " + receivedCount + " 条消息");
    }
    
    /**
     * 清理资源
     */
    private static void cleanupResources(Producer producer, Consumer consumer, String topicName) {
        try {
            // 取消订阅
            boolean unsubscribed = consumer.unsubscribe(topicName);
            if (unsubscribed) {
                System.out.println("消费者已取消订阅主题: " + topicName);
            }
            
            // 关闭消费者
            consumer.close();
            System.out.println("消费者已关闭");
            
            // 关闭生产者
            producer.close();
            System.out.println("生产者已关闭");
            
            // 删除主题（实际使用中，主题通常不会被轻易删除）
            // MessageQueueFactory.getInstance().getMessageQueue().deleteTopic(topicName);
            // System.out.println("主题已删除: " + topicName);
        } catch (Exception e) {
            System.err.println("清理资源时发生错误: " + e.getMessage());
            e.printStackTrace();
        }
    }
}